package com.atguigu.day08;

import com.atguigu.bean.WaterSensor;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.AllWindowedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.HashMap;

public class Flink03_ProcessFun_TopN {
    public static void main(String[] args) throws Exception {
        //1.获取流的执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //2.从端口读取数据
        SingleOutputStreamOperator<WaterSensor> waterSensorStream = env.socketTextStream("localhost", 9999)
                .map(new MapFunction<String, WaterSensor>() {
                    @Override
                    public WaterSensor map(String value) throws Exception {
                        String[] split = value.split(",");
                        return new WaterSensor(split[0], Long.parseLong(split[1]), Integer.parseInt(split[2]));
                    }
                }).assignTimestampsAndWatermarks(WatermarkStrategy.<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(0))
                        .withTimestampAssigner(new SerializableTimestampAssigner<WaterSensor>() {
                            @Override
                            public long extractTimestamp(WaterSensor element, long recordTimestamp) {
                                return element.getTs() * 1000;
                            }
                        })
                );

        //2.TODO 开启一个基于事件时间的滑动窗口 

        AllWindowedStream<WaterSensor, TimeWindow> windowedStream = waterSensorStream.windowAll(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(5)));

        // 例如，统计最近10秒钟内出现次数最多的两个水位，并且每5秒钟更新一次
        //1.先求每个vc出现了多少次  =》 WordCount
        //2.再对次数做倒序排序 获取前两个
        windowedStream.process(new MyTopNVcCount(3)).print();

        env.execute();
    }

    public static class MyTopNVcCount extends ProcessAllWindowFunction<WaterSensor, String, TimeWindow>{

        Integer topn;

        public MyTopNVcCount(Integer topn) {
            this.topn = topn;
        }

        @Override
        public void process(ProcessAllWindowFunction<WaterSensor, String, TimeWindow>.Context context, Iterable<WaterSensor> elements, Collector<String> out) throws Exception {
            //1.先求每个vc出现了多少次  =》 WordCount
            //创建HashMap用求不同水位的出现次数 k：vc  v：对应的次数
            HashMap<Integer, Integer> vcToCountMap = new HashMap<>();
            //2.遍历数据
            for (WaterSensor element : elements) {
                //3.求和
                if (vcToCountMap.containsKey(element.getVc())){
                    //不是第一条数据 所以要取出之前计算的结果然后累加
                    Integer lastCount = vcToCountMap.get(element.getVc());
                    vcToCountMap.put(element.getVc(), lastCount + 1);
                }else {
                    //第一条数据 那么直接将这个数据放入hashMap并计数为1
                    vcToCountMap.put(element.getVc(), 1);
                }
            }

            //3.将HashMap中的数据按照个数排序 因此HashMap没有提供排序的功能但是List集合有提供，所以将Map集合中的数据取出来放到List集合中
            ArrayList<Tuple2<Integer, Integer>> sortList = new ArrayList<>();
            for (Integer vc : vcToCountMap.keySet()) {
                sortList.add(Tuple2.of(vc, vcToCountMap.get(vc)));
            }

            //4.将list集合中的数据按照个数做倒序排序
            sortList.sort(new Comparator<Tuple2<Integer, Integer>>() {
                @Override
                public int compare(Tuple2<Integer, Integer> o1, Tuple2<Integer, Integer> o2) {
                    return o2.f1-o1.f1;
                }
            });

            if (sortList.size()<=topn){
                out.collect(sortList.toString());
            }else {
                //5.获取出现次数最高的两个vc
                for (int i = 0; i < topn; i++) {
                    out.collect(sortList.get(i)+"");
                }
            }

            String msg =
                    "窗口: [" + context.window().getStart() / 1000 + "," + context.window().getEnd() / 1000 + ") 一共有 "
                            + elements.spliterator().estimateSize() + "条数据 ";
            out.collect(msg);
        }
    }

}
